package mq

import (
	log "ac-common-go/glog"
	"ac-julink/tool/config"

	"github.com/streadway/amqp"
)

const exchangeType string = "direct"

type rabbitClient struct {
	clientPool *PoolManager
}

func (self *rabbitClient) InitClientPool() {
	mqurl := config.GetConfigValue("WGNS", "rabbitmq.url")
	queueName := config.GetConfigValue("WGNS", "rabbitmq.queue.status")
	exchange := config.GetConfigValue("WGNS", "rabbitmq.exchange.status")
	maxConn, ok := config.GetConfigIntValue("WGNS", "rabbitmq_maxconn")
	if !ok {
		maxConn = 3000
	}
	initConn, ok := config.GetConfigIntValue("WGNS", "rabbitmq_initconn")
	if !ok {
		initConn = 10
	}
	//	connTimeout, ok := config.GetConfigIntValue("WGNS", "rabbitmq_conntimeout")
	//	if !ok {
	//		connTimeout = 2
	//	}
	idleTimeout, ok := config.GetConfigIntValue("WGNS", "rabbitmq_idletimeout")
	if !ok {
		idleTimeout = 3600
	}
	retryNum, ok := config.GetConfigIntValue("WGNS", "rabbitmq_retrynum")
	if !ok {
		retryNum = 1
	}
	failMax, ok := config.GetConfigIntValue("WGNS", "rabbitmq_failmax")
	if !ok {
		failMax = 10000
	}
	self.clientPool = NewPoolManager(mqurl, maxConn, initConn, idleTimeout, retryNum, failMax,
		exchange, queueName)
	if self.clientPool == nil {
		log.Error("rabbit mq client pool init error!!!")
	}
}

func (self *rabbitClient) SendMessage(body string) error {
	log.Infof("start to send message to rabbit mq:%q", body)
	err := self.clientPool.SendMessage(body)
	if err != nil {
		log.Errorf("rabbit mq send error:%+v", err)
		return err
	}
	log.Info("rabbit mq send success")
	return nil
}

func (self *rabbitClient) sendMessageShort(body string) error {
	mqurl := "amqp://root:root@10.18.207.121:5672/vhost1"
	exchange := "testste"
	routingKey := "testss"
	connection, err := amqp.Dial(mqurl)
	if err != nil {
		log.Errorf("dial to %s error: %+v", mqurl, err)
		return err
	}
	defer connection.Close()
    log.Info("got Connection, getting Channel")
	channel, err := connection.Channel()
	if err != nil {
		log.Errorf("got channel error: %+v", err)
		return err
	}
    log.Infof("got Channel, declaring %q Exchange (%q)", exchangeType, exchange)
	if err := channel.ExchangeDeclare(
		exchange,     // name
		exchangeType, // type
		true,         // durable
		false,        // auto-deleted
		false,        // internal
		false,        // noWait
		nil,          // arguments
	); err != nil {
		log.Errorf("exchange declare: %+v", err)
		return err
	}
	channel.QueueDeclare(routingKey, false, false, false, false, nil)
	log.Infof("exchange declared, bind queue %q to exchange %q", routingKey, exchange)
	if err = channel.QueueBind(
		routingKey,
		routingKey,
		exchange,
		false,
		nil,
	); err != nil {
		log.Errorf("queue bind error: %q", err)
		return err
	}
	log.Infof("declared Exchange, publishing %dB body (%q)", len(body), body)
	if err = channel.Publish(
		exchange,   // publish to an exchange
		routingKey, // routing to 0 or more queues
		false,      // mandatory
		false,      // immediate
		amqp.Publishing{
			Headers:         amqp.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            []byte(body),
			DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
			Priority:        0,              // 0-9
			// a bunch of application/implementation-specific fields
		},
	); err != nil {
		log.Errorf("Exchange Publish: %+v", err)
		return err
	}
	return nil
}